package cn.doitedu.rtmk.demo2;


import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.math.BigDecimal;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/22
 * @Desc: 学大数据，上多易教育
 * 规则：
 * 静态画像条件：  gender=male，月平均订单数 > 10 的用户
 * 动态画像条件： 在规则上线后，历次发生的“添加购物车行为”中，添加的商品价格平均值>80.5元
 * 触发事件：    当他做 做下单行为时，给他推送消息
 **/
public class Demo2_Exercise {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);


        // 创建kafka 行为明细topic的source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd-events")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("doit43-8")
                .setClientIdPrefix("doitedu-c")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");
        SingleOutputStreamOperator<UserEvent> eventStream = streamSource.map(json -> JSON.parseObject(json, UserEvent.class));

        SingleOutputStreamOperator<String> hitMessages = eventStream
                .keyBy(UserEvent::getUser_id)
                .process(new KeyedProcessFunction<Long, UserEvent, String>() {
                    ValueState<BigDecimal> priceSumState;
                    ValueState<Integer> addCartCntState;
                    Table table;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 统计加购金额之和的状态
                        priceSumState = getRuntimeContext().getState(new ValueStateDescriptor<BigDecimal>("price_sum", BigDecimal.class));
                        // 统计加购次数的状态
                        addCartCntState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("add_cart_cnt", Integer.class));

                        // 创建hbase连接客户端
                        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
                        conf.set("hbase.zookeeper.quorum", "doitedu:2181");
                        Connection connection = ConnectionFactory.createConnection(conf);
                        table = connection.getTable(TableName.valueOf("user_profile"));

                    }

                    @Override
                    public void processElement(UserEvent userEvent, KeyedProcessFunction<Long, UserEvent, String>.Context ctx, Collector<String> out) throws Exception {

                        String eventId = userEvent.getEvent_id();
                        long userId = userEvent.getUser_id();

                        // 判断，收到的事件，是否是动态画像条件要统计的事件
                        if ("add_cart".equals(eventId)) {

                            /* *
                             * 累加商品的价格
                             */
                            // 取出状态中原来的和
                            BigDecimal oldSum = priceSumState.value() == null ? BigDecimal.ZERO : priceSumState.value();
                            // 获取本次加购的价格
                            String priceStr = userEvent.getProperties().get("price");
                            // 把字符串的价格转成bigDecimal
                            BigDecimal curPrice = BigDecimal.valueOf(Double.parseDouble(priceStr));
                            // 原来的和 加 本次的价格
                            BigDecimal added = oldSum.add(curPrice);
                            // 更新到状态中
                            priceSumState.update(added);


                            /**
                             * 累加加购的次数
                             */
                            // 取出状态中原来的累计值
                            int oldCnt = addCartCntState.value() == null ? 0 : addCartCntState.value();
                            addCartCntState.update(oldCnt + 1);


                        }

                        // 判断，收到的事件，是否是触发事件
                        if ("order_submit".equals(eventId)) {

                            // 先判断动态画像条件是否已经满足
                            if (priceSumState.value() != null
                                    && addCartCntState.value() != null
                                    && priceSumState.value().divide(BigDecimal.valueOf(addCartCntState.value())).compareTo(BigDecimal.valueOf(100.8)) > 0) {

                                // 再判断静态画像条件是否满足
                                Get get = new Get((userId + "").getBytes());
                                get.addColumn("f".getBytes(), "gender".getBytes());
                                get.addColumn("f".getBytes(), "tag103".getBytes());

                                Result result = table.get(get);
                                // 从result中提取查询到的标签值
                                byte[] genderBytes = result.getValue("f".getBytes(), "gender".getBytes());
                                byte[] tag103Bytes = result.getValue("f".getBytes(), "tag103".getBytes());

                                String gender = Bytes.toString(genderBytes);
                                String tag103Str = Bytes.toString(tag103Bytes); // "25.3"
                                double tag103 = Double.parseDouble(tag103Str);

                                // 判断查询到的该用户的画像标签值，是否满足规则中的条件
                                if (gender.equals("male") && tag103 > 10) {

                                    JSONObject message = new JSONObject();
                                    message.put("user_id", userId);
                                    message.put("hit_time", userEvent.getAction_time());
                                    message.put("rule_id", "rule-001");

                                    out.collect(message.toJSONString());

                                }


                            }


                        }


                    }
                });


        hitMessages.print();


        env.execute();

    }


}
